unit uSBDUnit_SyncObjects;
interface
uses SyncObjs, Windows, SysUtils, Contnrs, uSBD_BaseTypes;

const
  UsePropertyDefault = -2;
  WaitForever = -1;

type

ESemaphore = class abstract( TSBDUnit_WinAPIException)
  protected
    function StockMessage: string; override;
    function MsgPrefix: string;    override;
  end;

ESemaphoreConstructor = class( ESemaphore);
ESemaphoreRelease     = class( ESemaphore);
ESemaphoreWait        = class( ESemaphore);
ESemaphoreWierd       = class( ESemaphore);
ESemaphoreTimeOut     = class( ESemaphore);

TSBD_SemaphoreWaitResult = (rAcquiredSelf, rAcquiredMate, rTimeOut);

TSBD_Semaphore = class( TSynchroObject)
  private
    Fhandle: packed array of THandle;
    FTimeOut_ms: cardinal;
    FisSignalled: boolean;

    procedure BaseInit;
    procedure CheckConstruction( const Name: string);
    procedure ReleaseSemaphore;
    function  WaitSingle( Milliseconds: DWord): boolean;
    function  WaitForMultipleObjects( Milliseconds: DWORD; var Idx: integer): boolean;

  public
    constructor CreateSimple;

    constructor CreateEx(
      SecAttributesP: PSecurityAttributes;
      Initial, Maximum: integer; const Name: string);

    constructor CreateForeign( DesiredAccess: cardinal; const Name: string);
    destructor  Destroy; override;

    function  Mate   ( MateSemaphore: TSBD_Semaphore): integer;
    procedure Divorce( MateSemaphore: TSBD_Semaphore);

    procedure Acquire; override;
    procedure Release; override;
    function  AcquireEither( var MateIdx: integer; TimeOut: integer = UsePropertyDefault): TSBD_SemaphoreWaitResult;
    function  isSignalled: boolean;

    property  TimeOut_ms: cardinal   read FTimeOut_ms write FTimeOut_ms;
  end;

TTransThread_ObjectQueue = class( TObjectList)
  private
    FGate: TCriticalSection;
    FPops: TSBD_Semaphore;
    FisClosed: boolean;
    FMateIdx: integer;
    procedure SetTimeOut( Value_ms: integer);

  public
    constructor Create( MateSemaphore: TSBD_Semaphore);
    destructor  Destroy; override;

    procedure Push( Obj: TObject);
    function  WaitPop( TimeOut: integer = UsePropertyDefault): TObject;
    function  PeekPop: TObject;
    procedure Close;
    function  isClosed: boolean;
    procedure BrutalClose;         // Kills all the queued objects.

    property  TimeOut_ms: integer    write SetTimeOut;
  end;


implementation







{ TSBD_Semaphore }

procedure TSBD_Semaphore.BaseInit;
begin
FisSignalled := False;
SetLength( FHandle, 1);
FHandle[0] := 0;
FTimeOut_ms := INFINITE
end;


constructor TSBD_Semaphore.CreateEx(
  SecAttributesP: PSecurityAttributes;
  Initial, Maximum: integer; const Name: string);
begin
BaseInit;
FHandle[0] := windows.CreateSemaphore(
  SecAttributesP, Initial, Maximum, PChar( Name));
CheckConstruction( Name);
if (FHandle[0] <> 0) and (Name <> '') and
   (GetLastError = ERROR_ALREADY_EXISTS) then
  begin
  FHandle := nil;
  raise ESemaphoreConstructor.Create( Name, ERROR_ALREADY_EXISTS)
  end;
end;



constructor TSBD_Semaphore.CreateForeign(
  DesiredAccess: cardinal; const Name: string);
begin
BaseInit;
FHandle[0] := windows.OpenSemaphore( DesiredAccess, True, PChar( Name));
CheckConstruction( Name)
end;


constructor TSBD_Semaphore.CreateSimple;
begin
CreateEx( nil, 0, MaxInt, '')
end;



destructor TSBD_Semaphore.Destroy;
begin
if FHandle[0] <> 0 then
  begin
  windows.CloseHandle( FHandle[0]);
  FHandle[0] := 0
  end;
SetLength( FHandle, 0);
inherited
end;


function TSBD_Semaphore.AcquireEither(
  var MateIdx: integer; TimeOut: integer = UsePropertyDefault): TSBD_SemaphoreWaitResult;
var
  Idx: integer;
begin
if TimeOut = UsePropertyDefault then
  TimeOut := integer( FTimeOut_ms);
MateIdx := -1;
if Length( FHandle) <= 1 then
    begin
    if WaitSingle( cardinal( TimeOut)) then
        result := rAcquiredSelf
      else
        result := rTimeOut
    end
  else
    begin
    if WaitForMultipleObjects( cardinal( FTimeOut_ms), Idx) then
        begin
        MateIdx := Idx;
        if Idx = 0 then
            result  := rAcquiredSelf
          else
            result  := rAcquiredMate
        end
      else
        result := rTimeOut
    end
end;


procedure TSBD_Semaphore.Divorce( MateSemaphore: TSBD_Semaphore);
var
  Addend: cardinal;
  j, Idx, L: integer;
begin
Idx := -1;
L   := Length( FHandle);
if not assigned( MateSemaphore) or (Length( MateSemaphore.Fhandle) = 0) or
  (L < 2) or (MateSemaphore = self) then exit;
Addend := MateSemaphore.Fhandle[0];
Idx := -1;
for j := L - 1 downto 1 do
  begin
  if FHandle[j] <> Addend then continue;
  Idx := j;
  break
  end;
if Idx <> -1 then
  begin
  Dec( L);
  if Idx < L then
    Move( FHandle[Idx+1], FHandle[Idx], SizeOf( THandle) * (L - Idx));
  SetLength( FHandle, L)
  end
end;


function TSBD_Semaphore.isSignalled: boolean;
begin
result := FisSignalled
end;


function TSBD_Semaphore.Mate( MateSemaphore: TSBD_Semaphore): integer;
var
  Addend: cardinal;
  j: integer;
begin
result := -1;
if not assigned( MateSemaphore) or (Length( MateSemaphore.Fhandle) = 0) or
               (MateSemaphore = self) then exit;
Addend := MateSemaphore.Fhandle[0];
if (Addend = 0) or (Addend = ERROR_INVALID_HANDLE) then  exit;
for j := 0 to Length( FHandle) - 1 do
  begin
  if FHandle[j] <> Addend then continue;
  result := j;
  break
  end;
if result = -1 then
    begin
    result := Length( FHandle);
    SetLength( FHandle, result + 1);
    FHandle[ result] := Addend
    end
  else
    result := -1
end;


procedure TSBD_Semaphore.CheckConstruction( const Name: string);
var
  Handle: THandle;
begin
if Length( Fhandle) > 0 then
    Handle := FHandle[0]
  else
    Handle := 0;
if (Handle = 0) or (Handle = ERROR_INVALID_HANDLE) then
  raise ESemaphoreConstructor.Create
end;


procedure TSBD_Semaphore.Acquire;
var
  MateIdx: integer;
begin
if AcquireEither( MateIdx) = rTimeOut then
  raise ESemaphoreTimeOut.Create
end;

procedure TSBD_Semaphore.Release;
begin
ReleaseSemaphore;
FisSignalled := True
end;


procedure TSBD_Semaphore.ReleaseSemaphore;
begin
if integer( windows.ReleaseSemaphore( FHandle[0], 1, nil)) = 0 then
  raise ESemaphoreRelease.Create
end;


function TSBD_Semaphore.WaitForMultipleObjects(
  Milliseconds: DWORD; var Idx: integer): boolean;
var
  nCount: DWORD;
  bWaitAll: BOOL;
  dwWakeMask: DWORD;
  API_result: DWORD;
begin
Idx        := -1;
nCount     := Length( FHandle);
bWaitAll   := False;
dwWakeMask := 0;
API_result := windows.MsgWaitForMultipleObjects(
  nCount, FHandle[0], bWaitAll, Milliseconds, dwWakeMask);
case API_result of
  WAIT_TIMEOUT:  result := False;
  WAIT_FAILED :  raise ESemaphoreWait.Create('',ECode_NA);
  else
    begin
    if {(API_result >= WAIT_OBJECT_0) and }(API_result < WAIT_ABANDONED_0) and
       (API_result < (WAIT_OBJECT_0 + nCount)) then
        begin
        result := True;
        Idx :=  API_result - WAIT_OBJECT_0
        end
      else if (API_result >= WAIT_ABANDONED_0) and (API_result < WAIT_TIMEOUT)
          and (API_result < (WAIT_ABANDONED_0 + nCount)) then
        begin
        Idx :=  API_result - WAIT_ABANDONED_0;
        ESemaphoreWierd.Create('Semaphore abandoned',ECode_NA)
        end
      else
        ESemaphoreWierd.Create('Unrecognised result on semaphore wait',ECode_NA)
    end
  end
end;



function TSBD_Semaphore.WaitSingle( Milliseconds: DWord): boolean;
begin
case windows.WaitForSingleObject( FHandle[0], Milliseconds) of
  WAIT_ABANDONED: raise ESemaphoreWierd.Create('Semaphore abandoned',ECode_NA);
  WAIT_OBJECT_0 : result := True;
  WAIT_TIMEOUT  : result := False;
  WAIT_FAILED   : raise ESemaphoreWait.Create;
  else            raise ESemaphoreWierd.Create('Unrecognised result on semaphore wait');
  end;
end;



{ TTransThread_ObjectQueue }

constructor TTransThread_ObjectQueue.Create( MateSemaphore: TSBD_Semaphore);
begin
FisClosed := False;
inherited Create( True);
FGate := TCriticalSection.Create;
FPops := TSBD_Semaphore.CreateSimple;
FMateIdx := FPops.Mate( MateSemaphore)
end;


procedure TTransThread_ObjectQueue.BrutalClose;
var
  j: integer;
begin
FGate.Enter;
try
  FisClosed := True;
  OwnsObjects := False;
  for j := Count - 1 downto 0 do
    try Items[j].Free except end;
  Clear;
  OwnsObjects := True;
finally
  FGate.Leave
  end
end;

procedure TTransThread_ObjectQueue.Close;
begin
FGate.Enter;
try
  FisClosed := True
finally
  FGate.Leave
  end
end;


destructor TTransThread_ObjectQueue.Destroy;
begin
FGate.Enter;
try
  FisClosed := True;
  Clear
finally
  FGate.Leave
  end;
FGate.Free;
FPops.Free;
inherited
end;


function TTransThread_ObjectQueue.isClosed: boolean;
begin
result := FisClosed
end;


procedure TTransThread_ObjectQueue.Push( Obj: TObject);
begin
if not assigned( Obj) then exit;
if FisClosed then
  raise Exception.Create(' Object Queue is closed for producer.');
FGate.Enter;
try
  Add( Obj);
  FPops.Release
finally
  FGate.Leave
  end
end;


procedure TTransThread_ObjectQueue.SetTimeOut( Value_ms: integer);
begin
FPops.TimeOut_ms := Cardinal( Value_ms)
end;


function TTransThread_ObjectQueue.PeekPop: TObject;
begin
result := nil;
if Count = 0 then exit;
FGate.Enter;
try
  if Count > 0 then
    begin
    FPops.Acquire;
    result := Extract( Items[0])
    end
finally
  FGate.Leave
  end;
end;


function TTransThread_ObjectQueue.WaitPop( TimeOut: integer = UsePropertyDefault): TObject;
var
  MateIdx: integer;
begin
result := nil;
case FPops.AcquireEither( MateIdx, TimeOut) of
  rAcquiredSelf: begin
                 FGate.Enter;
                 try
                   if Count > 0 then
                     result := Extract( Items[0])
                 finally
                   FGate.Leave
                   end
                 end;

  rAcquiredMate: Close;

  rTimeOut     : Close;
  end
end;

{ ESemaphore }

function ESemaphore.MsgPrefix: string;
begin
result := ClassName + ': '
end;

function ESemaphore.StockMessage: string;
begin
result := 'Semaphore exception (Error code = %d)'
end;

end.
